package org.jeecg.common.es.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.GetIndexResponse;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.jeecg.common.es.CriteriaQueryProcessor;
import org.jeecg.common.es.EsIndexUtil;
import org.jeecg.common.es.entity.EsEntity;
import org.jeecg.common.exception.JeecgBootException;
import org.jeecg.common.util.DateUtils;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.document.Document;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.Criteria;
import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.UpdateQuery;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

import static org.elasticsearch.script.Script.DEFAULT_SCRIPT_LANG;

/**
 * @author wangzun1.vendor@sensetime.com
 * @Description:
 * @date 2022/8/10
 */
@Slf4j
@Service
public class EsServiceImpl extends AbstractEsService implements EsService, Serializable {


    public static final String PARAMS_POINT = "params.";
    public static final String CTX_SOURCE = "ctx._source.";
    @Override
    public <T extends EsEntity> void createIndex(List<T> entities) {
        // 将集合按照时间进行分区（yyyyMM）
        Map<String, List<T>> collect = entities.stream()
            .collect(Collectors.groupingBy(entity -> DateUtils.formatDate( entity.getSpiderTime(),"yyyyMM")));
        // 循环校验是否需要创建索引
        collect.forEach((key, value) -> {
            createIndexByEntity(value.get(0));
        });
    }

    /**
     * @see EsService#createIndex(EsEntity)
     */
    @Override
    public <T extends EsEntity> void createIndex(T entity) {
        // 创建索引
        createIndexByEntity(entity);
    }

    /**
     * @see EsService#save(EsEntity)
     */
    @Override
    public <T extends EsEntity> void save(T document) {
        // 防止save时新建索引（未设置别名，所以这里提前建好）
        createIndex(document);
        elasticsearchRestTemplate.save(document);
    }

    /**
     * @see EsService#save(List)
     */
    @Override
    public <T extends EsEntity> void save(List<T> documents) {
        // 防止save时新建索引（未设置别名，所以这里提前建好）
        createIndex(documents);
        elasticsearchRestTemplate.save(documents);
    }

    /**
     * @see EsService#search(CriteriaQuery, Class)
     */
    @Override
    public <T extends EsEntity> SearchHits<T> search(CriteriaQuery query, Class<T> clazz) {
        IndexCoordinates indexCoordinates = EsIndexUtil.aliasCoordinates(clazz);
        try {
            return elasticsearchRestTemplate.search(query, clazz, indexCoordinates);
        } catch (Exception e) {
            log.error(String.format("ES查询失败！query=%s errorInfo=%s", query, e.getMessage()), e);
            return null;
        }
    }

    /**
     * @see EsService#search(NativeSearchQueryBuilder, Class, boolean)
     */
    @Override
    public SearchHits search(NativeSearchQueryBuilder nativeSearchQueryBuilder, Class<? extends EsEntity> clazz,
        boolean createTimeDescSort) {
        nativeSearchQueryBuilder.withSearchType(SearchType.DFS_QUERY_THEN_FETCH);
        if (createTimeDescSort) {
            nativeSearchQueryBuilder
                .withSort(SortBuilders.fieldSort(EsEntity.Constants.SPIDER_TIME_FIELD).order(SortOrder.DESC));
        }
        IndexCoordinates indexCoordinates = EsIndexUtil.aliasCoordinates(clazz);
        try {
            return elasticsearchRestTemplate.search(nativeSearchQueryBuilder.build(), clazz, indexCoordinates);
        } catch (Exception e) {
            log.error(String.format("ES查询失败！nativeSearchQueryBuilder=%s errorInfo=%s", nativeSearchQueryBuilder,
                e.getMessage()), e);
            return null;
        }
    }

    /**
     * @see EsService#search(SearchRequest)
     */
    @Override
    public SearchResponse search(SearchRequest searchRequest) {
        try {
            return restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(String.format("ES查询失败！searchRequest=%s", searchRequest), e);
            throw new JeecgBootException("ES查询失败");
        } catch (ElasticsearchStatusException exception) {
            log.error("elastic search  status error, status={},message={}", exception.status(), exception.getMessage());
            return null;
        }
    }

    /**
     * @see EsService#deleteIndex(String)
     */
    @Override
    public void deleteIndex(String indexName) {
        try {
            if (EsIndexUtil.existIndex(getIndexResponse(indexName), indexName)) {
                restHighLevelClient.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
                refreshIndexCache(indexName);
            }
        } catch (Exception e) {
            log.error("删除索引出现异常", e);
            throw new JeecgBootException("ES删除索引失败！indexName=" + indexName);
        }
    }

    /**
     * @see EsService#getAllIndex(String)
     */
    @Override
    public String[] getAllIndex(String regular) throws IOException {
        GetIndexRequest getIndexRequest = new GetIndexRequest(regular);
        GetIndexResponse getIndexResponse = restHighLevelClient.indices().get(getIndexRequest, RequestOptions.DEFAULT);
        return getIndexResponse.getIndices();
    }

    /**
     * @see EsService#refreshByIndexName(String)
     */
    @Override
    public void refreshByIndexName(String indexName) {
        if (StringUtils.isNotBlank(indexName)) {
            RefreshRequest request = new RefreshRequest(indexName);
            try {
                restHighLevelClient.indices().refresh(request, RequestOptions.DEFAULT);
            } catch (Exception e) {
                log.error(String.format("刷新索引出现异常,indexName=%s", indexName), e);
            }
        }
    }

    /**
     * @see EsService#deleteByQuerySync(CriteriaQuery, Class, int)
     */
    @Override
    public boolean deleteByQuerySync(CriteriaQuery query, Class<? extends EsEntity> clazz, int batchSize) {
        IndexCoordinates indexCoordinates = EsIndexUtil.aliasCoordinates(clazz);
        /*// 此参数告诉 Elasticsearch把搜索结果的上下文保持5分钟
        if (query.getScrollTime() == null) {
            query.setScrollTime(Duration.ofMinutes(5));
        }*/
        query.setMaxResults(batchSize);
        elasticsearchRestTemplate.delete(query, clazz, indexCoordinates);
        SearchHits<?> queryResult = elasticsearchRestTemplate.search(query, clazz, indexCoordinates);
        return queryResult.getTotalHits() == 0L;
    }

    /**
     * @see EsService#deleteByIdSync(Collection, Class)
     */
    @Override
    public boolean deleteByIdSync(Collection<String> ids, Class<? extends EsEntity> clazz) {
        CriteriaQuery criteriaQuery = new CriteriaQuery(new Criteria().and("_id").in(ids));

        return deleteByQuerySync(criteriaQuery, clazz, ids.size());
    }

    /**
     * @see EsService#deleteByQueryAsync(CriteriaQuery, Class, ActionListener)
     */
    @Override
    public void deleteByQueryAsync(CriteriaQuery query, Class<? extends EsEntity> clazz,
        ActionListener<BulkByScrollResponse> listener) {
        QueryBuilder elasticsearchQuery = new CriteriaQueryProcessor().createQueryFromCriteria(query.getCriteria());
        String alias = EsIndexUtil.getRealAlias(clazz);
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(alias).setQuery(elasticsearchQuery)
            .setAbortOnVersionConflict(false).setRefresh(true);
        deleteByQueryRequest.setSlices(10);
        deleteByQueryRequest.setScroll(TimeValue.timeValueMillis(Duration.ofMinutes(10).toMillis()));
        try {
            restHighLevelClient.deleteByQueryAsync(deleteByQueryRequest, RequestOptions.DEFAULT, listener);
        } catch (Exception e) {
            log.error(String.format("异步删除索引出现异常,alias=%s", alias), e);
            throw new JeecgBootException("异步删除索引异常");
        }
    }

    /**
     * @see EsService#batchUpdateByQuery(Set, Class, Map)
     */
    @Override
    public void batchUpdateByQuery(Set<String> ids, Class<? extends EsEntity> clazz, Map<String, Object> updateValues) {
        UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest(EsIndexUtil.getRealAlias(clazz));
        StringBuilder scriptStringBuilder = new StringBuilder();
        Map<String, Object> params = new HashMap<>(10);
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.must(QueryBuilders.termsQuery("id", ids));
        updateByQueryRequest.setQuery(boolQueryBuilder);
        updateValues.forEach((key, value) -> {
            params.put(key, value);
            scriptStringBuilder.append(CTX_SOURCE).append(key).append("=").append(PARAMS_POINT).append(key).append(";");
        });
        updateByQueryRequest
            .setScript(new Script(ScriptType.INLINE, DEFAULT_SCRIPT_LANG, scriptStringBuilder.toString(), params));
        try {
            restHighLevelClient.updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error(String.format("ES批量更新记录操作失败，ids=%s  updateValues=%s realAlias=%s", ids, updateValues,
                EsIndexUtil.getRealAlias(clazz)), e);
            throw new JeecgBootException("ES批量更新失败");
        } finally {
            params.clear();
            updateValues.clear();
        }

    }

    /**
     * @see EsService#batchUpdateByQuery(List)
     */
    @Override
    public <T extends EsEntity> void batchUpdateByQuery(List<T> entities) throws IllegalAccessException {
        UpdateByQueryRequest updateByQueryRequest =
            new UpdateByQueryRequest(EsIndexUtil.getRealAlias(entities.get(0).getClass()));
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        Set<String> ids = entities.stream().map(EsEntity::getId).collect(Collectors.toSet());
        boolQueryBuilder.must(QueryBuilders.termsQuery("id", ids));
        try {
            for (T entity : entities) {
                /*根据实体映射更新记录*/
                doUpdateByQuery(entity, updateByQueryRequest);
            }
        } finally {
            entities.clear();
        }

    }

    /**
     * @see EsService#updateByQuery(String, Class, Map)
     */
    @Override
    public void updateByQuery(String id, Class<? extends EsEntity> clazz, Map<String, Object> updateValues) {
        UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest(EsIndexUtil.getRealAlias(clazz));
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.must(QueryBuilders.termQuery("id", id));
        updateByQueryRequest.setQuery(boolQueryBuilder);
        StringBuilder scriptStringBuilder = new StringBuilder();
        Map<String, Object> params = new HashMap<>(10);
        updateValues.forEach((key, value) -> {
            params.put(key, value);
            scriptStringBuilder.append(CTX_SOURCE).append(key).append("=").append(PARAMS_POINT).append(key).append(";");
        });
        updateByQueryRequest
            .setScript(new Script(ScriptType.INLINE, DEFAULT_SCRIPT_LANG, scriptStringBuilder.toString(), params));
        try {
            restHighLevelClient.updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error(String.format("ES更新记录操作失败，id=%s  updateValues=%s realAlias=%s", id, updateValues,
                EsIndexUtil.getRealAlias(clazz)), e);
            throw new JeecgBootException("ES操作失败");
        } finally {
            params.clear();
            updateValues.clear();
        }
    }

    /**
     * @see EsService#updateByQuery(T)
     */
    @Override
    public <T extends EsEntity> void updateByQuery(T entity) throws IllegalAccessException {
        UpdateByQueryRequest updateByQueryRequest =
            new UpdateByQueryRequest(EsIndexUtil.getRealAlias(entity.getClass()));
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.must(QueryBuilders.termQuery("id", entity.getId()));
        updateByQueryRequest.setQuery(boolQueryBuilder);
        /*根据实体映射更新记录*/
        doUpdateByQuery(entity, updateByQueryRequest);
    }

    /**
     * @see EsService#update(String, String, Class, Map)
     */
    @Override
    public void update(String id, String realIndexName, Class<? extends EsEntity> clazz,
        Map<String, Object> updateValues) {
        Document document = Document.create();
        for (String field : updateValues.keySet()) {
            document.put(field, updateValues.get(field));
        }
        IndexCoordinates indexCoordinates = IndexCoordinates.of(realIndexName);
        UpdateQuery updateQuery = UpdateQuery.builder(id).withDocument(document).build();
        try {
            elasticsearchRestTemplate.update(updateQuery, indexCoordinates);
        } catch (Exception e) {
            log.error(String.format("ES更新记录操作失败，id=%s  updateValues=%s realAlias=%s", id, updateValues,
                EsIndexUtil.getRealAlias(clazz)), e);
            throw new JeecgBootException("ES操作失败");
        }

    }

    /**
     * @see EsService#update(T)
     */
    @Override
    public <T extends EsEntity> void update(T entity) throws IllegalAccessException {
        Document document = Document.create();
        List<Field> fieldList = new ArrayList<>();
        Class tempClass = entity.getClass();
        // 递归获取所有的属性
        while (tempClass != null) {
            fieldList.addAll(Arrays.asList(tempClass.getDeclaredFields()));
            // 得到父类,然后赋给自己
            tempClass = tempClass.getSuperclass();
        }
        for (Field field : fieldList) {
            field.setAccessible(true);
            Object value = field.get(entity);
            if (value != null) {
                String name = field.getName();
                document.put(name, value);
            }
        }
        String realIndexName = getRelIndexName(entity.getClass(), DateUtils.formatDate(entity.getSpiderTime(),"yyyyMM"));
        IndexCoordinates indexCoordinates = IndexCoordinates.of(realIndexName);
        UpdateQuery updateQuery = UpdateQuery.builder(entity.getId()).withDocument(document).build();
        try {
            elasticsearchRestTemplate.update(updateQuery, indexCoordinates);
        } catch (Exception e) {
            log.error(String.format("ES更新记录操作失败，entity=%s  ", entity), e);
            throw new JeecgBootException("ES更新记录操作失败");
        } finally {
            document.clear();
            fieldList.clear();
        }
    }

    /**
     * 根据实体映射更新记录
     *
     * @param entity 实体
     * @param updateByQueryRequest es更新请求实例
     * @param <T> 泛型
     * @throws IllegalAccessException
     */
    private <T extends EsEntity> void doUpdateByQuery(T entity, UpdateByQueryRequest updateByQueryRequest)
        throws IllegalAccessException {
        StringBuilder scriptStringBuilder = new StringBuilder();
        List<Field> fieldList = new ArrayList<>();
        Class tempClass = entity.getClass();
        // 递归获取所有的属性
        while (tempClass != null) {
            fieldList.addAll(Arrays.asList(tempClass.getDeclaredFields()));
            // 得到父类,然后赋给自己
            tempClass = tempClass.getSuperclass();
        }
        // 动态参数集合
        Map<String, Object> params = new HashMap<>(10);
        for (Field field : fieldList) {
            field.setAccessible(true);
            Object value = field.get(entity);
            if (value != null) {
                String name = field.getName();
                params.put(name, value);
                scriptStringBuilder.append(CTX_SOURCE).append(name).append("=").append(PARAMS_POINT).append(name)
                    .append(";");
            }
        }
        updateByQueryRequest
            .setScript(new Script(ScriptType.INLINE, DEFAULT_SCRIPT_LANG, scriptStringBuilder.toString(), params));
        try {
            restHighLevelClient.updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            log.error(String.format("ES更新记录操作失败，entity=%s", entity), e);
            throw new JeecgBootException("ES更新记录操作失败");
        } finally {
            fieldList.clear();
            params.clear();
        }
    }
}
